package com.atguigu.flink.chapter6;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/22 9:08
 */
public class Flink03_Project_ByChannel {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 把渠道和行为做key, 然后分组, 最后求和
        env
          .addSource(new MyAppBehavior())
          .map(behavior -> Tuple2.of(behavior.getBehavior() + "_" + behavior.getChannel(), 1L))
          .returns(Types.TUPLE(Types.STRING, Types.LONG))
          .keyBy(t -> t.f0)
          .sum(1)
          .print();
        env.execute();


    }

    public static class MyAppBehavior implements SourceFunction<MarketingUserBehavior>{
        private Random random = new Random();
        private List<String> channels = Arrays.asList("huawei", "xiaomi", "apple", "baidu", "qq", "vivo", "oppo");
        private List<String> behaviors = Arrays.asList("download", "install", "update", "uninstall");

        private volatile boolean isRunning = true;

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (isRunning){
                Long userId = (long)random.nextInt(10000) + 1;
                String behavior = behaviors.get(random.nextInt(behaviors.size()));
                String channel = channels.get(random.nextInt(channels.size()));
                Long timestamp = System.currentTimeMillis();
                ctx.collect(new MarketingUserBehavior(userId, behavior, channel, timestamp));
                Thread.sleep(2000);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
